package com.aotain.nyx.port;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

import com.aotain.apollo.mongodb.MongoImportTool;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.common.mongo.DataColumn;
import com.aotain.nyx.ipstat.IPStatTuple;
import com.aotain.nyx.statis.AbnStatisTuple;

/**
 * 端口异常天统计
     *@Title:
     *@Description:
     *@Author:Turk
     *@Since:2017年8月15日
     *@Version:1.1.0
 */
public class PortAbnStatis  implements FlatMapFunction<Tuple2<String,IPStatTuple>, AbnStatisTuple>{

	/**
	 *
	 */
	    
	private static final long serialVersionUID = 4481684808536814242L;

	@Override
	public void flatMap(Tuple2<String, IPStatTuple> value,
			Collector<AbnStatisTuple> out) throws Exception {
		
		IPStatTuple tuple = value.f1;
		

		String dip = tuple.getDestIP();
		
		Date current = new Date();
		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMdd");
		String strDate = sdf.format(current);
		
		String dPort = tuple.getDestPort();
		
		//String rowkey = String.format("%s_%s_%s", strDateD, dip, port);
		
		String tableName = ApolloProperties.SDS_PORT_STAT_DAY;
		
		List<DataColumn> filter = new ArrayList<DataColumn>();
		filter.add(new DataColumn("REPORTDATE",Long.parseLong(strDate)));
		filter.add(new DataColumn("IP",dip));
		
		List<DataColumn> saverow = new ArrayList<DataColumn>();
		saverow.add(new DataColumn("REPORTDATE", Long.parseLong(strDate)));
		saverow.add(new DataColumn("IP", dip));
		saverow.add(new DataColumn("PORT", dPort));
		
		List<DataColumn> incrow = new ArrayList<DataColumn>();
		incrow.add(new DataColumn("PORTNUM", tuple.getSessionNum()));
		
		MongoImportTool.getInstance().UpdateRowData(tableName, filter, saverow);
		MongoImportTool.getInstance().IncRowData(tableName, filter, incrow);
		
		
	}

}

    